package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.beans.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.utils.DateFormatUtil;
import com.atguigu.gmall.realtime.utils.MyClickhouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2023/4/23
 * sku粒度下单业务过程聚合统计
 *      维度：sku
 *      度量：统计原始金额、活动减免金额、优惠券减免金额和订单金额
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、clickhouse、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 * 开发流程
 *      -环境准备
 *      -检查点相关的设置
 *      -从kafka的主题中(下单事实表)读取数据
 *      -对读取的数据进行类型转换并过滤掉空消息   jsonStr->jsonObj
 *      -去重
 *          为什么会产生空消息和重复数据？
 *              因为我们是从下单事实表中读取数据的，下单事实表是将业务数据库中的订单明细表、订单表、订单明细活动表、
 *              订单明细优惠券表进行关联，订单明细表是主表，它和订单表在关联的时候使用的是内连接，在明细活动以及
 *              明细优惠券进行关联的时候，使用的是左外连接，如果左表数据先到，会产生如下数据
 *              左表      null    +I
 *              左表      null    -D
 *              左表      右表     +I
 *              这样的动态表数据保存到kafka主题中后，kakfa主题会接收到3条消息
 *              左表      null
 *              null
 *              左表      右表
 *          -对于空消息的处理
 *              如果使用的kafka连接器读取空消息的话，会将空消息自动过滤掉
 *              如果使用的API的方式读取空消息的话，Flink默认提供的SimpleStringSchema是处理不了空消息的，需要
 *              在封装创建FlinkKafkaConsumer工具方法中重写反序列化方法
 *          -去重思路1：状态 + 定时器
 *              当第一条数据来的时候，将数据放到状态中，并注册一个5秒之后执行的定时器；
 *              当第二条数据来的时候，从状态中取出第一条数据并进行时间的比较，将时间大的保存到状态中;
 *              当定时器被触发之后，将状态中数据传递到下游
 *              优点：传递的数据量小
 *              弊端：每来一条数据，都要过5秒后才会传递到下游，影响整个实时数仓的时效性
 *          -去重思路2：状态  +  抵消
 *              当第一条数据来的时候，将数据放到状态中，并传递到下游；
 *              当第二条数据来的时候，将状态中影响到度量值的属性进行取反，并传递到下游
 *              将第二条数据向下游传递
 *      -将流中的类型转换为要统计的实体类对象
 *      -指定Watermark以及提取事件时间字段
 *      -按照统计的维度skuId进行分组
 *      -开窗
 *      -聚合计算
 *      -维度关联
 *          基本的维度关联
 *              PhoenixUtil----List<T> queryList(sql,Class)
 *                  从Phoenix表中查询数据并将查询的结果封装为T类型的对象放到List集合中
 *              DimUtil----JSONObject getDimInfoNoCache(String tableName,Tuple2...params)
 *                  底层还是调用PhoenixUtil中的queryList
 *              在DwsTradeSkuOrderWindow应用，通过map算子，对聚合之后的流中数据进行处理，补充维度
 *                  根据流中的对象获取要关联的维度主键
 *                  根据维度的主键获取维度对象
 *                  将维度对象的属性补充到流中对象上
 *          优化1：旁路缓存
 *              思路：先从缓存中查询维度数据，如果缓存中找到了对象的维度，那直接将缓存中的数据作为方法的返回值进行返回；
 *                  如果在缓存中没有找到对应的维度，发送请求到Phoenix表中将维度数据查询出来，并放到缓存中，方便下次查询使用
 *              缓存产品选型
 *                  状态      性能高     维护性差
 *                  redis     性能也不错 维护性好    √
 *              redis相关的配置
 *                  key:    dim:维度表表名:主键
 *                  type:   string
 *                  ttl:    1day
 *                  注意：如果维度数据发生了变化  需要将缓存的数据清掉
 *          优化2：异步IO
 *              为什么使用异步IO?
 *                  要想提升对流中数据的处理能力，可以调大并行度，但是更多的并行度意味着需要更多的硬件资源，所以不能通过调大
 *                  并行度的方式提升处理能力
 *                  在单个并行度上，如果使用map算子处理流中数据，使用同步的方式，处理完一个元素之后，再对下一个元素进行处理，
 *                  在和外部系统进行交互的时候，等待的时间比较长，影响到整个实时项目的时效性。
 *                  所以我们使用异步的方式对单个并行度上的数据进行处理，同时处理多个元素，等待的时间会被发送请求给稀释
 *              如果发送异步请求？
 *                  AsyncDataStream.[un]orderedWait(
 *                      流,
 *                      异步请求 implements AsyncFunction,
 *                          调用支持异步操作的API
 *                          通过线程的方式----通过双重校验锁解决单例设计模式懒汉式线程安全的问题
 *                      超时时间,
 *                      时间单位
 *                  )
 *
 *                  class DimAsyncFunction extends RichAsyncFunction{
 *                      open:初始化线程池对象
 *                      asyncInvoke:
 *                          开启线程   √
 *                          根据流中对象获取要关联的维度的主键  ×  ---抽象
 *                          根据维度的主键获取维度对象 √
 *                          将维度对象的属性补充到流中的对象上  ×  ---抽象
 *                          向下游传递  √
 *                  }
 *                  模板方法设计模式：
 *                      在父类中定义完成某一个功能的核心算法的骨架(步骤)，但是有些步骤在父类中没办法实现，
 *                      可以延迟到子类中去实现，在不改变父类核心算法骨架的前提下，每一个子类都可以有自己不同的实现。
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);

        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka的下单事实表主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        //TODO 4.过滤掉空消息  并对流中的数据类型进行转换  jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
            new ProcessFunction<String, JSONObject>() {
                @Override
                public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                    if (StringUtils.isNotEmpty(jsonStr)) {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        out.collect(jsonObj);
                    }
                }
            }
        );
        // {"create_time":"2023-04-11 09:02:44","sku_num":"1","activity_rule_id":"1","split_original_amount":"6999.0000",
        // "split_coupon_amount":"0.0","sku_id":"1","date_id":"2023-04-11","user_id":"328","province_id":"2",
        // "activity_id":"1","sku_name":"小米12","id":"14262295","order_id":"55565","split_activity_amount":"500.0",
        // "split_total_amount":"6499.0","ts":"1682211764"}
        // jsonObjDS.print(">>>");

        //TODO 5.按照唯一键(订单明细id)进行分组  目的：去重
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS
            = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));

        //TODO 6.去重
        /*//6.1 状态 + 定时器,  这种方案需要过5s后，才会将数据传递到下游，对时效性有影响
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSNObject, JSONObject>() {
                private 2ValueState<JSONObject> lastValueState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor
                        = new ValueStateDescriptor<JSONObject>("lastValueState",JSONObject.class);
                    lastValueState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject lastValue = lastValueState.value();
                    if(lastValue == null){
                        //如果状态中的内容为空，说明是第一条数据，没有重复，将这条数据放到状态中，并注册定时器
                        lastValueState.update(jsonObj);
                        long currentProcessingTime = ctx.timerService().currentProcessingTime();
                        ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                    }else {
                        //伪代码
                        Long ts1 = jsonObj.getLong("ts1");
                        Long ts2 = lastValue.getLong("ts2");
                        if(ts1 >= ts2){
                            lastValueState.update(jsonObj);
                        }
                    }
                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                    //从状态中获取数据
                    JSONObject jsonObj = lastValueState.value();
                    out.collect(jsonObj);
                    //清空状态
                    lastValueState.clear();
                }
            }
        );*/
        //6.2 状态 + 抵消
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastValueState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor
                        = new ValueStateDescriptor<JSONObject>("lastValueState", JSONObject.class);
                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                    lastValueState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject lastValue = lastValueState.value();
                    if (lastValue != null) {
                        String splitOriginalAmount = lastValue.getString("split_original_amount");
                        String splitTotalAmount = lastValue.getString("split_total_amount");
                        String splitCouponAmount = lastValue.getString("split_coupon_amount");
                        String activityAmount = lastValue.getString("split_activity_amount");
                        lastValue.put("split_original_amount", "-" + splitOriginalAmount);
                        lastValue.put("split_total_amount", "-" + splitTotalAmount);
                        lastValue.put("split_coupon_amount", "-" + splitCouponAmount);
                        lastValue.put("split_activity_amount", "-" + activityAmount);
                        out.collect(lastValue);
                    }
                    out.collect(jsonObj);
                    lastValueState.update(jsonObj);
                }
            }
        );
        // distinctDS.print(">>>>");
        //TODO 7.对流中数据类型进行转换    jsonObj->统计的实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> orderBeanDS = distinctDS.map(
            new MapFunction<JSONObject, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                    // {"create_time":"2023-04-11 09:02:44","sku_num":"1","activity_rule_id":"1","split_original_amount":"6999.0000",
                    // "split_coupon_amount":"0.0","sku_id":"1","date_id":"2023-04-11","user_id":"328","province_id":"2",
                    // "activity_id":"1","sku_name":"小米12","id":"14262295","order_id":"55565","split_activity_amount":"500.0",
                    // "split_total_amount":"6499.0","ts":"1682211764"}
                    String skuId = jsonObj.getString("sku_id");
                    Long ts = jsonObj.getLong("ts") * 1000;
                    String splitOriginalAmount = jsonObj.getString("split_original_amount");
                    String splitTotalAmount = jsonObj.getString("split_total_amount");
                    String splitCouponAmount = jsonObj.getString("split_coupon_amount");
                    String splitActivityAmount = jsonObj.getString("split_activity_amount");
                    TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                        .skuId(skuId)
                        .originalAmount(new BigDecimal(splitOriginalAmount))
                        .orderAmount(new BigDecimal(splitTotalAmount))
                        .couponAmount(new BigDecimal(splitCouponAmount))
                        .activityAmount(new BigDecimal(splitActivityAmount))
                        .ts(ts)
                        .build();
                    return orderBean;
                }
            }
        );
        // orderBeanDS.print(">>>");

        //TODO 8.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = orderBeanDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<TradeSkuOrderBean>forMonotonousTimestamps()
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                        @Override
                        public long extractTimestamp(TradeSkuOrderBean orderBean, long recordTimestamp) {
                            return orderBean.getTs();
                        }
                    }
                )
        );

        //TODO 9.按照sku维度进行分组    目的：聚合
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);

        //TODO 10.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS
            = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 11.聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                    value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                    value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                    value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                    value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                    return value1;
                }
            },
            new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                @Override
                public void apply(String groupId, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                    String stt = DateFormatUtil.toYmdHms(window.getStart());
                    String edt = DateFormatUtil.toYmdHms(window.getEnd());
                    for (TradeSkuOrderBean orderBean : input) {
                        orderBean.setStt(stt);
                        orderBean.setEdt(edt);
                        orderBean.setTs(System.currentTimeMillis());
                        out.collect(orderBean);
                    }
                }
            }
        );
        // TradeSkuOrderBean(stt=2023-04-23 10:33:30, edt=2023-04-23 10:33:40,
        // trademarkId=null, trademarkName=null, category1Id=null, category1Name=null,
        // category2Id=null, category2Name=null, category3Id=null, category3Name=null,
        // skuId=31, skuName=null, spuId=null, spuName=null, originalAmount=552.0000,
        // activityAmount=0.0, couponAmount=0.0, orderAmount=552.0, ts=1682217220101)

        // reduceDS.print(">>>>");
        //TODO 12.关联sku维度
        /*SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new MapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                    //根据流中对象获取要关联的维度的主键
                    String skuId = orderBean.getSkuId();
                    //根据维度的主键到对应的维度表中获取维度对象
                    JSONObject dimJsonObj = DimUtil.getDimInfo("dim_sku_info", skuId);
                    //将维度对象的属性补充到流中的对象上
                    orderBean.setSkuName(dimJsonObj.getString("SKU_NAME"));
                    orderBean.setSpuId(dimJsonObj.getString("SPU_ID"));
                    orderBean.setCategory3Id(dimJsonObj.getString("CATEGORY3_ID"));
                    orderBean.setTrademarkId(dimJsonObj.getString("TM_ID"));
                    return orderBean;
                }
            }
        );
        */
        /*//将异步I/O操作应用于DataStream作为DataStream的一次转换操作
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            //发送异步请求  进行维度关联 实现分发请求的 AsyncFunction
            new AsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                @Override
                public void asyncInvoke(TradeSkuOrderBean orderBean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {
                    //开启线程
                    ThreadPoolExecutor poolExecutor = ThreadPoolUtil.getInstance();
                    poolExecutor.submit(new Runnable() {
                        @Override
                        public void run() {
                            //根据流中的对象获取要关联的维度的主键
                            String skuId = orderBean.getSkuId();
                            //根据维度的主键获取维度对象
                            JSONObject dimJsonObj = DimUtil.getDimInfo("dim_sku_info", skuId);
                            //将维度对象的属性 补充到流中的对象
                            orderBean.setSkuName(dimJsonObj.getString("SKU_NAME"));
                            orderBean.setSpuId(dimJsonObj.getString("SPU_ID"));
                            orderBean.setCategory3Id(dimJsonObj.getString("CATEGORY3_ID"));
                            orderBean.setTrademarkId(dimJsonObj.getString("TM_ID"));
                            //获取数据库交互的结果并发送给ResultFuture的回调函数，向下游传递数据
                            resultFuture.complete(Collections.singleton(orderBean));
                        }
                    });
                }
            },
            60,
            TimeUnit.SECONDS
        );*/
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            //发送异步请求  进行维度关联 实现分发请求的 AsyncFunction
            new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                    orderBean.setSkuName(dimJsonObj.getString("SKU_NAME"));
                    orderBean.setSpuId(dimJsonObj.getString("SPU_ID"));
                    orderBean.setCategory3Id(dimJsonObj.getString("CATEGORY3_ID"));
                    orderBean.setTrademarkId(dimJsonObj.getString("TM_ID"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            },
            60,
            TimeUnit.SECONDS
        );
        // withSkuInfoDS.print(">>>>");

        //TODO 13.关联spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
            withSkuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setSpuName(dimInfoJsonObj.getString("SPU_NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSpuId();
                }
            },
            60, TimeUnit.SECONDS
        );

        // withSpuInfoDS.print(">>>");

        //TODO 14.关联tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
            withSpuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setTrademarkName(dimInfoJsonObj.getString("TM_NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getTrademarkId();
                }
            },
            60, TimeUnit.SECONDS
        );
        // withTmDS.print(">>>");

        //TODO 15.关联category3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory3Stream = AsyncDataStream.unorderedWait(
            withTmDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3".toUpperCase()) {
                @Override
                public void join(TradeSkuOrderBean javaBean,JSONObject jsonObj)  {
                    javaBean.setCategory3Name(jsonObj.getString("name".toUpperCase()));
                    javaBean.setCategory2Id(jsonObj.getString("category2_id".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory3Id();
                }
            },
            60, TimeUnit.SECONDS
        );

        //TODO 16.关联category2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory2Stream = AsyncDataStream.unorderedWait(
            withCategory3Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2".toUpperCase()) {
                @Override
                public void join(TradeSkuOrderBean javaBean,JSONObject jsonObj)  {
                    javaBean.setCategory2Name(jsonObj.getString("name".toUpperCase()));
                    javaBean.setCategory1Id(jsonObj.getString("category1_id".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory2Id();
                }
            },
            60, TimeUnit.SECONDS
        );

        //TODO 17.关联category1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory1Stream = AsyncDataStream.unorderedWait(
            withCategory2Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1".toUpperCase()) {
                @Override
                public void join(TradeSkuOrderBean javaBean,JSONObject jsonObj)  {
                    javaBean.setCategory1Name(jsonObj.getString("name".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory1Id();
                }
            },
            60, TimeUnit.SECONDS
        );

        //TODO 18.将关联的结果写到CK中
        withCategory1Stream.print(">>>>");
        withCategory1Stream.addSink(
            MyClickhouseUtil.getSinkFunction("insert into dws_trade_sku_order_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
        );

        env.execute();
    }
}
